package com.key.w8.http1.connection.connectionImpl;

import com.key.w8.common.execute.ChainExecuteHandler;
import com.key.w8.common.execute.ExecuteHandlerChain;
import com.key.w8.http1.connection.Connection;
import com.key.w8.http1.entity.HttpObject;
import com.key.w8.http1.entity.HttpRequest;
import com.key.w8.http1.entity.HttpResponse;
import com.key.w8.http1.handler.codec.Http1Codec;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.HttpHeaderNames;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 当请求太大就会有模拟请求，服务器需要针对模拟请求给出对应的响应，而客户端需要针对响应做出处理，如抛异常
 * 当响应太大了会有模拟响应，客户端需要针对模拟响应做处理，抛异常
 * 异常都是在客户端抛，分别是请求太大和响应太大
 *
 * @Author: k8
 * @CreateTime: 2025-02-24
 * @Version: 1.0
 */
@Slf4j
public class DefaultHttp1Connection implements Connection {
    private Set<ErrorListener> errorListeners = new ConcurrentSkipListSet<>();

    private ExecuteHandlerChain executeChain = new ExecuteHandlerChain();
    private ExecuteHandlerChain erroExecuteChain = new ExecuteHandlerChain();
    private List<CloseListener> closeListeners = new LinkedList<>();
    private List<CreatedListener> createdListeners = new LinkedList<>();
    private List<WriteRequestListener> writeRequestListeners = new LinkedList<>();

    private volatile State state = State.INITIAL;
    private Lock stateChangeLock = new ReentrantLock();
    protected Http1Codec http1Codec;

    @Override
    public Connection addCloseListener(CloseListener closeListener) {
        this.closeListeners.add(closeListener);
        return this;
    }

    @Override
    public Connection addCreatedListener(CreatedListener createdListener) {
        this.createdListeners.add(createdListener);
        return this;
    }

    public void setHttp1Codec(Http1Codec http1Codec) {
        this.http1Codec = http1Codec;
    }




    @Override
    public Connection addErrorListener(ErrorListener errorListener){
        assert errorListener != null;
        this.errorListeners.add(errorListener);
        return this;
    }
    public Connection addChainExecuteHandler(ChainExecuteHandler handler){
        assert handler != null;
        this.executeChain.addExecuteHandler(handler);
        return this;
    }
    @Override
    public Connection addChainErrorHandler(ChainExecuteHandler handler){
        assert handler != null;
        this.erroExecuteChain.addExecuteHandler(handler);
        return this;
    }
    @Override
    public State state() {
        return state;
    }

    @Override
    public void state(State state) {
        try {
            stateChangeLock.lock();
            if (state.value <= this.state.value) return;
            this.state = state;
        }finally {
            stateChangeLock.unlock();
        }
        //closeListener中可能会再次调用state，为了避免死锁，需要将其挪出来
        if (state == State.CLOSED){
            closeListeners.forEach(closeListener -> {
                closeListener.onClose(this);
            });
        }else if (state == State.CREATED){
            createdListeners.forEach(createdListener -> {
                createdListener.onCreated(this);
            });
        }
    }
    private boolean check(){
        if (this.state == State.CREATED) return true;
        return false;
    }


    /**
     * 不需要获取写状态的
     * @param httpRequest
     */
    @Override
    public void onRequestRead(HttpRequest httpRequest) {
        try {
            executeChain.invoke(httpRequest,this);
        } catch (Throwable e) {
            log.debug("------------------------Invoke request error.---"+this.getClass().getName());
            throw new RuntimeException(e);
        }
    }

    @Override
    public void onExecuteError(HttpObject httpObject,Throwable throwable) {
        try {
            errorListeners.forEach(errorListener -> {
                errorListener.error(throwable);
            });
            erroExecuteChain.invoke(httpObject,new ErrorEntity(throwable,this));
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 具体的交给handler进行处理
     * @param response
     */
    @Override
    public void onResponseRead(HttpResponse response) {
        try {
            executeChain.invoke(response,null);
        } catch (Throwable e) {
            log.debug("------------------------Invoke response error.---"+this.getClass().getName());
            throw new RuntimeException(e);
        }
    }



    @Override
    public void writeRequest(HttpRequest httpRequest, CompletableFuture<Void> requestFuture) {
        if (requestFuture == null || requestFuture.isDone()){
            writeRequest(httpRequest);
            return;
        }
        preWriteRequest(httpRequest);
        ChannelPromise promise = getPromise().addListener((future) -> {
                    if (future.isSuccess()) {
                        requestFuture.complete(null);
                    } else if (future.cause() != null) {
                        requestFuture.completeExceptionally(future.cause());
                    } else {
                        requestFuture.cancel(true);
                    }
                });
        http1Codec.writeRequest(promise,httpRequest);
    }
    public ChannelPromise getPromise(){
        return http1Codec.getPromise();
    }
    @Override
    public void writeRequest(HttpRequest httpRequest) {
        preWriteRequest(httpRequest);
        http1Codec.writeRequest(httpRequest);
    }

    @Override
    public void writeRequest(HttpRequest httpRequest, ChannelPromise promise) {
        preWriteRequest(httpRequest);
        http1Codec.writeRequest(promise,httpRequest);
    }

    @Override
    public void writeRequest(byte[] data, ChannelPromise promise) {
        http1Codec.writeRequestContent(data,"",promise);
    }

    @Override
    public void writeRequest(byte[] data, CompletableFuture<Void> requestFuture) {
        if (requestFuture == null || requestFuture.isDone()){
            writeRequest(data);
            return;
        }
        ChannelPromise promise = getPromise().addListener((future) -> {
                    if (future.isSuccess()) {
                        requestFuture.complete(null);
                    } else if (future.cause() != null) {
                        requestFuture.completeExceptionally(future.cause());
                    } else {
                        requestFuture.cancel(true);
                    }
                });
        http1Codec.writeRequestContent(promise,data, "");
    }

    @Override
    public void writeRequest(byte[] data) {
        http1Codec.writeResponseContent(data);
    }



    @Override
    public void writeResponse(HttpResponse httpResponse, CompletableFuture<Void> responseFuture) {
        if (responseFuture == null || responseFuture.isDone()){
            writeResponse(httpResponse);
            return;
        }
        ChannelPromise promise = getPromise().addListener((future) -> {
            if (future.isSuccess()) {
                responseFuture.complete(null);
            } else if (future.cause() != null) {
                responseFuture.completeExceptionally(future.cause());
            } else {
                responseFuture.cancel(true);
            }
        });
        http1Codec.writeResponse(promise,httpResponse);
    }

    @Override
    public void writeResponse(HttpResponse httpResponse, ChannelPromise promise) {
        http1Codec.writeResponse(promise,httpResponse);
    }

    @Override
    public void writeResponse(HttpResponse httpResponse) {
        http1Codec.writeResponse(httpResponse);
    }

    @Override
    public void writeResponse(byte[] data, CompletableFuture<Void> responseFuture) {
        if (responseFuture == null || responseFuture.isDone()){
            writeResponse(data);
            return;
        }
        ChannelPromise promise = getPromise().addListener((future) -> {
            try {
                future.get();
                responseFuture.complete(null);
            }catch (Throwable throwable){
                responseFuture.completeExceptionally(throwable);
            }
        });
        http1Codec.writeResponseContent(promise,data);
    }

    @Override
    public void writeResponse(byte[] data, ChannelPromise promise) {
        http1Codec.writeResponseContent(promise,data);
    }

    @Override
    public void writeResponse(byte[] data) {
        http1Codec.writeResponseContent(data);
    }

    @Override
    public void close() throws IOException {
        if (state() == State.CLOSED) return;
        state(State.CLOSED);
    }

    @Override
    public void shutDown() throws IOException {
        close();
    }

    @Override
    public void tryClose() throws IOException {
        close();
    }
    class ErrorEntity{
        Throwable throwable;
        Connection connection;

        public ErrorEntity(Throwable throwable, Connection connection) {
            assert throwable != null;
            assert connection != null;
            this.throwable = throwable;
            this.connection = connection;
        }

        public Throwable getThrowable() {
            return throwable;
        }

        public Connection getConnection() {
            return connection;
        }
    }

    @Override
    public ByteBufAllocator alloc() {
        return http1Codec.alloc();
    }

    @Override
    public void addWriteRequestListener(WriteRequestListener writeRequestListener) {
        assert writeRequestListener != null;
        writeRequestListeners.add(writeRequestListener);
    }

    private void preWriteRequest(HttpRequest httpRequest){
        for (WriteRequestListener writeRequestListener : writeRequestListeners) {
            writeRequestListener.preWriteRequest(httpRequest);
        }
    }
}
